package com.haotian.kafka;

import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.kafka.streams.state.HostInfo;

abstract class KafkaBasic {

	private boolean isConnected = true;

	public KafkaBasic(_KafkaConfig config) throws UnknownHostException, IOException {
		// 创建独立的 检测连接

		final List<HostInfo> hosts = config.getHosts();

		final List<Socket> clients = new ArrayList<>();

		for (HostInfo hostInfo : hosts) {
				clients.add(new Socket(hostInfo.host(), hostInfo.port()));
		}

		// 每秒检测一次所有ip和地址

		Timer timer = new Timer();

		timer.schedule(new TimerTask() {

			@Override
			public void run() {
				for (Socket socket : clients) {
					try {
						socket.sendUrgentData(0xFF);
						isConnected = true;
					} catch (IOException e) {
						isConnected = false;
					}
				}
			}
		}, 0, 1000);
	}

	/**
	 * 集群的情况下 断开单个机器 也会出现 false和true 来回变化的情况 所有应该判断false 是否为连续的出现
	 * 
	 * @return
	 */
	public boolean isConnected() {
		return this.isConnected;
	}
}
